package com.ksyun.campus.metaserver.mq.producer;

import com.alibaba.fastjson2.JSON;
import com.ksyun.campus.metaserver.mq.event.SyncMetaDataEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import static com.ksyun.campus.metaserver.common.constant.MqConstant.SYNC_META_DATA_KEY;
import static com.ksyun.campus.metaserver.common.constant.MqConstant.SYNC_META_DATA_TOPIC;

/**
 * 主节点向从节点发送同步元数据消息生产者
 */

@Slf4j
@Component
@RequiredArgsConstructor
public class SyncMetaDataProducer {
    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 消息发送，自定义 Topic
     */
    public void send(SyncMetaDataEvent syncMetaDataEvent) {
        // consumer不能直接简单create 副本可能会变
        // consumer 消费重试（需要抛异常提醒broker）
        log.info("开始发送同步消息");
        Message<?> message = MessageBuilder
                .withPayload(syncMetaDataEvent)
                .setHeader(MessageConst.PROPERTY_KEYS, SYNC_META_DATA_KEY)
                .build();
        try {
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(SYNC_META_DATA_TOPIC, message, SYNC_META_DATA_KEY);
            log.info("消息发送结果：{}，消息ID：{}，消息体：{}", sendResult.getSendStatus(), sendResult.getMsgId(), JSON.toJSONString(syncMetaDataEvent));
        } catch (Throwable ex) {
            throw new RuntimeException(String.format("消息发送失败，[消息体]: %s", JSON.toJSONString(syncMetaDataEvent)), ex);
        }
    }
}
